package com.king.first.app.uv;

import com.king.first.bean.UserBehavior;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

import java.util.HashMap;
import java.util.HashSet;

public class Flink02_Project_UV {

    public static void main1(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        DataStreamSource<String> dataStreamSource = env.readTextFile("input/UserBehavior.csv");
        dataStreamSource.flatMap((String line, Collector<Tuple2<String, Long>> out) -> {
                    String[] split = line.split(",");
                    UserBehavior behavior = new UserBehavior(
                            Long.valueOf(split[0]),
                            Long.valueOf(split[1]),
                            Integer.valueOf(split[2]),
                            split[3],
                            Long.valueOf(split[4]));
                    if ("pv".equals(behavior.getBehavior())) {
                        out.collect(Tuple2.of("uv" + behavior.getUserId(), 1L));
                    }
                }).returns(Types.TUPLE(Types.STRING, Types.LONG))
                .keyBy(t -> t.f0)
                .process(new KeyedProcessFunction<String, Tuple2<String, Long>, HashMap<String, Long>>() {
                    private HashMap<String, Long> idsMap = new HashMap();
                    @Override
                    public void processElement(Tuple2<String, Long> stringLongTuple2, KeyedProcessFunction<String, Tuple2<String, Long>, HashMap<String, Long>>.Context context, Collector<HashMap<String, Long>> collector) throws Exception {
                        String f0 = stringLongTuple2.f0;
                        Long count = idsMap.getOrDefault(f0, 0L);
                        count += 1;
                        idsMap.put(f0, count);
                        collector.collect(idsMap);
                    }
                })
                .print();

        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }


    public static void main2(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env
                .readTextFile("input/UserBehavior.csv")
                .flatMap((String line, Collector<Tuple2<String, Long>> out) -> {
                    String[] split = line.split(",");
                    UserBehavior behavior = new UserBehavior(
                            Long.valueOf(split[0]),
                            Long.valueOf(split[1]),
                            Integer.valueOf(split[2]),
                            split[3],
                            Long.valueOf(split[4]));
                    if ("pv".equals(behavior.getBehavior())) {
                        out.collect(Tuple2.of("uv" + behavior.getUserId(), behavior.getUserId()));
                    }
                }).returns(Types.TUPLE(Types.STRING, Types.LONG))
                .keyBy(t -> t.f0)
                .process(new KeyedProcessFunction<String, Tuple2<String, Long>, HashMap<String, Long>>() {
                    HashSet<Long> userIds = new HashSet<>();
                    HashMap<String, Long> idsMap = new HashMap<>();
                    @Override
                    public void processElement(Tuple2<String, Long> value, Context ctx, Collector<HashMap<String, Long>> out) throws Exception {
                        userIds.add(value.f1);
                        Long aLong = idsMap.getOrDefault(value.f0, 0L);
                        aLong = aLong + 1;
                        idsMap.put(value.f0, aLong);
                        out.collect(idsMap);
                    }
                })
                .print("uv");

        env.execute();

    }
}
